package com.arch.flink.window.watermark;

import org.apache.commons.lang3.time.FastDateFormat;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssigner;
import org.apache.flink.api.common.eventtime.TimestampAssignerSupplier;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.assigners.SlidingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.concurrent.TimeUnit;

/**
 * @author pizhihui
 * @date 2024-03-11 14:23
 */
public class Demo3EventTimeBox {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        env.setParallelism(1);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        DataStreamSource<String> source = env.addSource(new TestSource());

        // strategy
        WatermarkStrategy<Tuple2<String, Long>> strategy = WatermarkStrategy
                .<Tuple2<String, Long>>forBoundedOutOfOrderness(Duration.ofSeconds(5))
                .withTimestampAssigner(new SerializableTimestampAssigner<Tuple2<String, Long>>() {
                    @Override
                    public long extractTimestamp(Tuple2<String, Long> element, long recordTimestamp) {
                        return element.f1;
                    }
                });

        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOne = source.map(new MapFunction<String, Tuple2<String, Long>>() {
            @Override
            public Tuple2<String, Long> map(String s) throws Exception {
                String[] split = s.split(",");
//                        for (String s1 : split) {
//                            collector.collect(Tuple2.of(s1, s2));
//                        }
                // collector.collect(Tuple2.of(split[0], Long.parseLong(split[1])));
                return Tuple2.of(split[0], Long.parseLong(split[1]));
            }
        });

        // assign
        SingleOutputStreamOperator<Tuple2<String, Long>> wordAndOneWatermark =
                wordAndOne.assignTimestampsAndWatermarks(strategy);

        SingleOutputStreamOperator<Tuple2<String, Integer>> res = wordAndOneWatermark
                // keyby
                .keyBy(new KeySelector<Tuple2<String, Long>, String>() {
                    @Override
                    public String getKey(Tuple2<String, Long> stringIntegerTuple2) throws Exception {
                        return stringIntegerTuple2.f0;
                    }
                })

                // window
//                 .window(SlidingProcessingTimeWindows.of(Time.seconds(10), Time.seconds(5)))
                .window(SlidingEventTimeWindows.of(Time.seconds(10), Time.seconds(5)))

                // process
                .process(new ProcessWindowFunction<Tuple2<String, Long>, Tuple2<String, Integer>, String, TimeWindow>() {

                    FastDateFormat dataFormat = FastDateFormat.getInstance("HH:mm:ss");

                    @Override
                    public void process(String o,
                                        ProcessWindowFunction<Tuple2<String, Long>,
                                                Tuple2<String, Integer>, String, TimeWindow>.Context context,
                                        Iterable<Tuple2<String, Long>> elements,
                                        Collector<Tuple2<String, Integer>> out) throws Exception {


                        long now = System.currentTimeMillis();
                        long l = context.currentProcessingTime();
                        long l1 = context.currentWatermark();
                        long start = context.window().getStart();
                        long end = context.window().getEnd();


                        System.out.println(
                                "当前时间=" + dataFormat.format(now) + "," +
                                        "processing time=" + dataFormat.format(l) + "," +
                                        "watermark=" + dataFormat.format(l1) + "," +
                                        "start=" + dataFormat.format(start) + ",end=" + dataFormat.format(end));

                        int sum = 0;
                        for (Tuple2<String, Long> element : elements) {
                            System.out.println("dddd:" + element.f0 + "," + dataFormat.format(element.f1));
                            sum += 1;
                        }
                        out.collect(Tuple2.of(o, sum));
                    }
                });


        res.print().setParallelism(1);

        env.execute("Socket3 WordCount");


    }

    static class TestSource implements SourceFunction<String> {

        FastDateFormat dateFormat = FastDateFormat.getInstance("HH:mm:ss");

        @Override
        public void run(SourceContext<String> ctx) throws Exception {
            // 控制大约在 10 秒的倍数的时间点发送事件
            String currTime = String.valueOf(System.currentTimeMillis());

            while (Integer.valueOf(currTime.substring(currTime.length() - 4)) > 100) {
                currTime = String.valueOf(System.currentTimeMillis());
                continue;
            }
            System.out.println("开始发送事件的时间：" + dateFormat.format(System.currentTimeMillis()));
            // 第 13 秒发送两个事件
            TimeUnit.SECONDS.sleep(3);
            ctx.collect("hadoop," + System.currentTimeMillis());
            // 产生了一个事件，但是由于网络原因，事件没有发送
            //  ctx.collect("hadoop," + System.currentTimeMillis());
            String event = "hadoop," + System.currentTimeMillis();

            // 第 16 秒发送一个事件
            TimeUnit.SECONDS.sleep(3);
            ctx.collect("hadoop," + System.currentTimeMillis());

            // ***** 显示延迟发送的
            // 第 19 秒的时候发送13的事情
            TimeUnit.SECONDS.sleep(3);
            ctx.collect(event);

            TimeUnit.SECONDS.sleep(300);
        }

        @Override
        public void cancel() {

        }
    }

}
